package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 针对硬盘上的消息进行管理
 */
public class MessageFileManager {

    //用内部类来表示该队列的统计信息
    //一般内部类优先考虑静态内部类，不依赖于外部，解耦更彻底
    static public class Stat{
        //总消息数目
        public int totalCount;

        //有效消息数目
        public int validCount;
    }

    public void init(){

    }

    /**
     * 约定消息文件所在的目录和文件名
     */

    //用来获取到指定队列对应的消息文件所在的路径
    private String getQueueDir(String queueName){
        return "./data/" + queueName;
    }

    //用来获取该队列的消息数据文件路径（二进制文件）
    private String getQueueDataPath(String queueName){
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //用来获取该队列的消息统计文件的路径
    private String getQueueStatPath(String queueName){
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    private Stat readStat(String queueName){
        Stat stat = new Stat();
        try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName, Stat stat){
        //OutputStream 打开文件，默认将原文件清空，新数据覆盖旧数据
        try(OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))){
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        //1.创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            //不存在就创建
            boolean ok = baseDir.mkdirs();
            if(!ok){
                throw new IOException("创建目录失败！ baseDir = " + baseDir.getAbsolutePath());
            }
        }
        //2.创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            boolean ok = queueDataFile.createNewFile();
            if(!ok){
                throw new IOException("创建文件失败！ queueDataFile = " + queueDataFile.getAbsolutePath());
            }
        }
        //3.创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            boolean ok = queueStatFile.createNewFile();
            if(!ok){
                throw new IOException("创建文件失败！ queueStatFile = " + queueStatFile.getAbsolutePath());
            }
        }
        Stat stat = new Stat();
        stat.validCount = 0;
        stat.totalCount = 0;
        writeStat(queueName, stat);
    }

    //删除队列的目录和文件
    public void destoryQueueFiles(String queueName) throws IOException {
        //先删除文件再删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!ok1 || !ok2 || !ok3){
            throw new IOException("删除队列目录和文件失败！baseDir = " + baseDir.getAbsolutePath());
        }
    }

     //检查队列的目录和文件是否存在
    public boolean checkFilesExists(String queueName){
        //判定数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            return false;
        }
        return true;
    }

    //把新的消息放到队列对应的文件中
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.检查要写入的队列对应文件是否存在
        if(!checkFilesExists(queue.getName())){
            throw new MqException("[MessageFileManager] 队列对应的文件不存在！ queueName = " + queue.getName());
        }
        //2.把 Message 对象进行序列化
        byte[] messageBinary = BinaryTool.toBytes(message);
        synchronized (queue){
            //3.获取当前队列数据文件长度，用来计算该 Message 对象的 offsetBeg 和 offsetEnd
            // 将该 Message 对象写入到数据文件末尾
            // 此时该 Message 对象的 offsetBeg = 当前文件长度 + 4，offsetEnd = offsetBeg + message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            //4.写入消息到数据文件，追加写入到数据文件末尾
            try(OutputStream outputStream = new FileOutputStream(queueDataFile, true)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    //先写入当前消息的长度, 占 4 个字节
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            //5.更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.validCount += 1;
            stat.totalCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 删除消息（逻辑删除，isValid = 0）
     * @param queue
     * @param message
     * 1.
     */
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue){
            try(RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")){
                //1.先从文件中读取对应的 Message
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                //2.把二进制反序列化成 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                //3.将 isValid 设置为 无效
                diskMessage.setIsValid((byte) 0x0);
                //4.重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }
            //更新统计文件
            Stat stat = readStat(queue.getName());
            if(stat.validCount > 0){
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 从文件中读取所有的消息内容加载内存中
     * 在程序启动的时候调用，此时服务器还不能申请请求，不涉及多线程
     * @param queueName
     * @return
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)){
                //记录当前文件光标
                long currentOffset = 0;
                //一个文件中包含很多消息，使用循环
                while (true){
                    //1.读取当前消息长度
                    int messageSize = dataInputStream.readInt();
                    //2.按照上述长度读取内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if(messageSize != actualSize){
                        //计划长度与实际长度不匹配，文件有问题
                        throw new MqException("[MessageFileManager] 文件格式错误！ queueName = " + queueName);
                    }
                    //3. 将读到的二进制数据反序列化回 Message 对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    //4. 判定消息是否有效
                    if(message.getIsValid() != 0x1){
                        //无效，跳过，更新 offset
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    //5.有效数据，将 Message 对象添加到链表中，更新 offsetBeg 和 offsetEnd
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            }catch (EOFException e){
                //这个 catch 不是处理异常，而是处理正常业务逻辑，文件督导末尾会被 readInt 抛出该异常
                System.out.println("[MessageFileManager] 恢复 Message 数据完成！");
            }
        }
        return messages;
    }

    //检查当前队列消息的数据文件是否需要GC
    public boolean checkGC(String queueName){
        //总消息超过 2000 ，并且有效消息 小于总数的一半 就进行GC
        Stat stat = readStat(queueName);
        if(stat.totalCount > 2000 && (double)stat.totalCount / (double) stat.validCount < 0.5){
            return true;
        }
        return false;
    }

    //垃圾回收后有效消息在新的数据文件中
    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    /**
     * 使用复制算法来完成垃圾回收
     * 1. 创建新的文件，名字为 queue_data_new.txt
     * 2. 将之前数据文件中的有效消息读出来，写入到新的文件中
     * 3. 删除旧的文件，再将新的文件名改回到 queue_data.txt
     * 4. 更新统计消息文件
     * @param queue
     */
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        synchronized (queue){
            //统计耗时
            long gcBeg = System.currentTimeMillis();

            //1.创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常情况下文件不应该存在，如果存在说明上次 gc 了一半程序崩溃
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在！" +
                        "queueName = " + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败！ queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }
            //2. 从旧的文件中，读取所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            //3. 把有效消息，写入到新文件中
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for(Message message : messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //4. 删除旧的文件数据，把新的重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败！ queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }
            //把 queue_data_new.txt => queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败！ queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }
            //5.更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕！ queueName = " + queue.getName() + ", time = "
            + (gcEnd - gcBeg) + "ms");
        }
    }
}
